package cn.ctyun.datax.plugin.reader.httpreader;

import cn.ctyun.datax.plugin.reader.httpreader.constant.HttpConstant;
import cn.ctyun.datax.plugin.reader.httpreader.constant.HttpReaderErrorCode;
import cn.ctyun.datax.plugin.reader.httpreader.constant.KeyConstant;
import cn.ctyun.datax.plugin.reader.httpreader.util.HttpUtil;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * HTTP reader插件
 *
 * @author cuijianpeng
 * @date 2020/05/09 13:45
 */
public class HttpReader extends Reader {

    public static class Job extends Reader.Job {

        private static final Logger LOG = LoggerFactory.getLogger(Job.class);
        private Configuration originalConfig = null;

        @Override
        public void init() {
            this.originalConfig = super.getPluginJobConf();
            final List<String> urlList = originalConfig.getList(KeyConstant.URL, String.class);
            if(Objects.isNull(urlList) || urlList.isEmpty()){
                throw DataXException.asDataXException(HttpReaderErrorCode.URL_NULL, "请修改后重试！");
            }
            originalConfig.getNecessaryValue(KeyConstant.HTTP_METHOD, HttpReaderErrorCode.HTTP_METHOD_NULL);
            originalConfig.getNecessaryValue(KeyConstant.CODE, HttpReaderErrorCode.SUCCESS_CODE_NULL);
            originalConfig.getNecessaryValue(KeyConstant.CODE_JSON_PATH, HttpReaderErrorCode.CODE_JSON_PATH_NULL);
            originalConfig.getNecessaryValue(KeyConstant.DATA_JSON_PATH, HttpReaderErrorCode.DATA_JSON_PATH_NULL);
            final Long delayTime = this.originalConfig.getLong(KeyConstant.DELAY_TIME);
            if(Objects.isNull(delayTime) || delayTime < HttpConstant.DEFAULT_DELAY_TIME){
                this.originalConfig.set(KeyConstant.DELAY_TIME, HttpConstant.DEFAULT_DELAY_TIME);
                LOG.debug(String.format("delayTime使用默认值 %s ms", HttpConstant.DEFAULT_DELAY_TIME));
            }
        }

        @Override
        public void preCheck() { }

        /**
         * 考虑到接口容易访问失败，目前只允许分片数为1
         *
         * @param adviceNumber 未使用
         * @return 分片数
         */
        @Override
        public List<Configuration> split(int adviceNumber) {
            List<Configuration> splitConfigs = new ArrayList<>();
            splitConfigs.add(originalConfig);
            return splitConfigs;
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

    }

    public static class Task extends Reader.Task {

        private static final Logger LOG = LoggerFactory.getLogger(HttpReader.Task.class);
        private Configuration readerSliceConfig;

        @Override
        public void init() {
            this.readerSliceConfig = super.getPluginJobConf();
        }

        @Override
        public void startRead(RecordSender sender) {
            List<Configuration> splitConfigs = HttpUtil.splitJob(readerSliceConfig);
            for (Configuration splitConfig : splitConfigs) {
                final String response = HttpUtil.urlRequest(splitConfig);
                final List<Object> parseData = HttpUtil.parseResponse(splitConfig, response);
                HttpUtil.transportRecord(splitConfig, parseData, sender, super.getTaskPluginCollector());
            }
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

    }

}
